// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#include <nebula/types/field.h>
#include <turbo/log/logging.h>
#include <nebula/numeric/decimal.h>
#include <nebula/types/util.h>
#include <algorithm>
#include <climits>
#include <cstddef>
#include <iterator>
#include <limits>
#include <memory>
#include <mutex>
#include <ostream>
#include <sstream>  // IWYU pragma: keep
#include <string>
#include <unordered_map>
#include <unordered_set>
#include <utility>
#include <vector>

#include <nebula/core/array.h>
#include <nebula/core/compare.h>
#include <nebula/core/record_batch.h>
#include <nebula/types/fundamental.h>
#include <turbo/utility/status.h>
#include <nebula/core/table.h>
#include <turbo/base/checked_cast.h>
#include <nebula/numeric/decimal.h>
#include <nebula/util/hash_util.h>
#include <nebula/util/hashing.h>
#include <nebula/util/key_value_metadata.h>
#include <turbo/log/logging.h>
#include <nebula/util/range.h>
#include <nebula/util/string.h>

#include <nebula/util/vector.h>
#include <nebula/types/type.h>

namespace nebula {

    namespace {
        turbo::Result<std::shared_ptr<DataType>> MergeTypes(std::shared_ptr<DataType> promoted_type,
                                                            std::shared_ptr<DataType> other_type,
                                                            const Field::MergeOptions &options);

        turbo::Result<std::shared_ptr<DataType>> WidenDecimals(
                const std::shared_ptr<DataType> &promoted_type,
                const std::shared_ptr<DataType> &other_type, const Field::MergeOptions &options) {
            const auto &left = turbo::checked_cast<const DecimalType &>(*promoted_type);
            const auto &right = turbo::checked_cast<const DecimalType &>(*other_type);
            if (!options.promote_numeric_width && left.bit_width() != right.bit_width()) {
                return turbo::failed_precondition_error(
                        "Cannot promote decimal128 to decimal256 without promote_numeric_width=true");
            }
            const int32_t max_scale = std::max<int32_t>(left.scale(), right.scale());
            const int32_t common_precision =
                    std::max<int32_t>(left.precision() + max_scale - left.scale(),
                                      right.precision() + max_scale - right.scale());
            if (left.id() == Type::DECIMAL256 || right.id() == Type::DECIMAL256 ||
                common_precision > BasicDecimal128::kMaxPrecision) {
                return DecimalType::create(Type::DECIMAL256, common_precision, max_scale);
            }
            return DecimalType::create(Type::DECIMAL128, common_precision, max_scale);
        }

        // Merge numeric types based on options. Returns nullptr for non-numeric types.
        turbo::Result<std::shared_ptr<DataType>> MaybeMergeNumericTypes(
                std::shared_ptr<DataType> promoted_type, std::shared_ptr<DataType> other_type,
                const Field::MergeOptions &options) {
            bool promoted = false;
            if (options.promote_decimal_to_float) {
                if (is_decimal(promoted_type->id()) && is_floating(other_type->id())) {
                    promoted_type = other_type;
                    promoted = true;
                } else if (is_floating(promoted_type->id()) && is_decimal(other_type->id())) {
                    other_type = promoted_type;
                    promoted = true;
                }
            }

            if (options.promote_integer_to_decimal &&
                ((is_decimal(promoted_type->id()) && is_integer(other_type->id())) ||
                 (is_decimal(other_type->id()) && is_integer(promoted_type->id())))) {
                if (is_integer(promoted_type->id()) && is_decimal(other_type->id())) {
                    // Other type is always the int
                    promoted_type.swap(other_type);
                }
                TURBO_MOVE_OR_RAISE(const int32_t precision,
                                    MaxDecimalDigitsForInteger(other_type->id()));
                TURBO_MOVE_OR_RAISE(const auto promoted_decimal,
                                    DecimalType::create(promoted_type->id(), precision, 0));
                TURBO_MOVE_OR_RAISE(promoted_type,
                                    WidenDecimals(promoted_type, promoted_decimal, options));
                return promoted_type;
            }

            if (options.promote_decimal && is_decimal(promoted_type->id()) &&
                is_decimal(other_type->id())) {
                TURBO_MOVE_OR_RAISE(promoted_type,
                                    WidenDecimals(promoted_type, other_type, options));
                return promoted_type;
            }

            if (options.promote_integer_sign && ((is_unsigned_integer(promoted_type->id()) &&
                                                  is_signed_integer(other_type->id())) ||
                                                 (is_signed_integer(promoted_type->id()) &&
                                                  is_unsigned_integer(other_type->id())))) {
                if (is_signed_integer(promoted_type->id()) && is_unsigned_integer(other_type->id())) {
                    // Other type is always the signed int
                    promoted_type.swap(other_type);
                }

                if (!options.promote_numeric_width &&
                    bit_width(promoted_type->id()) < bit_width(other_type->id())) {
                    return turbo::failed_precondition_error(
                            "Cannot widen signed integers without promote_numeric_width=true");
                }
                int max_width =
                        std::max<int>(bit_width(promoted_type->id()), bit_width(other_type->id()));

                // If the unsigned one is bigger or equal to the signed one, we need another bit
                if (bit_width(promoted_type->id()) >= bit_width(other_type->id())) {
                    ++max_width;
                }

                if (max_width > 32) {
                    promoted_type = int64();
                } else if (max_width > 16) {
                    promoted_type = int32();
                } else if (max_width > 8) {
                    promoted_type = int16();
                } else {
                    promoted_type = int8();
                }
                return promoted_type;
            }

            if (options.promote_integer_to_float &&
                ((is_floating(promoted_type->id()) && is_integer(other_type->id())) ||
                 (is_integer(promoted_type->id()) && is_floating(other_type->id())))) {
                if (is_integer(promoted_type->id()) && is_floating(other_type->id())) {
                    // Other type is always the int
                    promoted_type.swap(other_type);
                }

                const int int_width = bit_width(other_type->id());
                promoted = true;
                if (int_width <= 8) {
                    other_type = float16();
                } else if (int_width <= 16) {
                    other_type = float32();
                } else {
                    other_type = float64();
                }

                if (!options.promote_numeric_width &&
                    bit_width(promoted_type->id()) != bit_width(other_type->id())) {
                    return turbo::failed_precondition_error("Cannot widen float without promote_numeric_width=true");
                }
            }

            if (options.promote_numeric_width) {
                const int max_width =
                        std::max<int>(bit_width(promoted_type->id()), bit_width(other_type->id()));
                if (is_floating(promoted_type->id()) && is_floating(other_type->id())) {
                    promoted = true;
                    if (max_width >= 64) {
                        promoted_type = float64();
                    } else if (max_width >= 32) {
                        promoted_type = float32();
                    } else {
                        promoted_type = float16();
                    }
                } else if (is_signed_integer(promoted_type->id()) &&
                           is_signed_integer(other_type->id())) {
                    promoted = true;
                    if (max_width >= 64) {
                        promoted_type = int64();
                    } else if (max_width >= 32) {
                        promoted_type = int32();
                    } else if (max_width >= 16) {
                        promoted_type = int16();
                    } else {
                        promoted_type = int8();
                    }
                } else if (is_unsigned_integer(promoted_type->id()) &&
                           is_unsigned_integer(other_type->id())) {
                    promoted = true;
                    if (max_width >= 64) {
                        promoted_type = uint64();
                    } else if (max_width >= 32) {
                        promoted_type = uint32();
                    } else if (max_width >= 16) {
                        promoted_type = uint16();
                    } else {
                        promoted_type = uint8();
                    }
                }
            }

            return promoted ? promoted_type : nullptr;
        }


        // Merge two dictionary types, or else give an error.
        turbo::Result<std::shared_ptr<DataType>> MergeDictionaryTypes(
                const std::shared_ptr<DataType> &promoted_type,
                const std::shared_ptr<DataType> &other_type, const Field::MergeOptions &options) {
            const auto &left = turbo::checked_cast<const DictionaryType &>(*promoted_type);
            const auto &right = turbo::checked_cast<const DictionaryType &>(*other_type);
            if (!options.promote_dictionary_ordered && left.ordered() != right.ordered()) {
                return turbo::failed_precondition_error(
                        "Cannot merge ordered and unordered dictionary unless "
                        "promote_dictionary_ordered=true");
            }
            Field::MergeOptions index_options = options;
            index_options.promote_integer_sign = true;
            index_options.promote_numeric_width = true;
            TURBO_MOVE_OR_RAISE(
                    auto indices,
                    MaybeMergeNumericTypes(left.index_type(), right.index_type(), index_options));
            TURBO_MOVE_OR_RAISE(auto values,
                                MergeTypes(left.get_value_type(), right.get_value_type(), options));
            auto ordered = left.ordered() && right.ordered();
            if (indices && values) {
                return dictionary(indices, values, ordered);
            } else if (values) {
                return turbo::failed_precondition_error("Could not merge dictionary index types");
            }
            return turbo::failed_precondition_error("Could not merge dictionary value types");
        }

        // Merge temporal types based on options. Returns nullptr for non-temporal types.
        turbo::Result<std::shared_ptr<DataType>> MaybeMergeTemporalTypes(
                const std::shared_ptr<DataType> &promoted_type,
                const std::shared_ptr<DataType> &other_type, const Field::MergeOptions &options) {
            if (options.promote_temporal_unit) {
                if (promoted_type->id() == Type::DATE32 && other_type->id() == Type::DATE64) {
                    return date64();
                }
                if (promoted_type->id() == Type::DATE64 && other_type->id() == Type::DATE32) {
                    return date64();
                }

                if (promoted_type->id() == Type::DURATION && other_type->id() == Type::DURATION) {
                    const auto &left = turbo::checked_cast<const DurationType &>(*promoted_type);
                    const auto &right = turbo::checked_cast<const DurationType &>(*other_type);
                    return duration(std::max(left.unit(), right.unit()));
                }

                if (is_time(promoted_type->id()) && is_time(other_type->id())) {
                    const auto &left = turbo::checked_cast<const TimeType &>(*promoted_type);
                    const auto &right = turbo::checked_cast<const TimeType &>(*other_type);
                    const auto unit = std::max(left.unit(), right.unit());
                    if (unit == TimeUnit::MICRO || unit == TimeUnit::NANO) {
                        return time64(unit);
                    }
                    return time32(unit);
                }
            }

            if (promoted_type->id() == Type::TIMESTAMP && other_type->id() == Type::TIMESTAMP) {
                const auto &left = turbo::checked_cast<const TimestampType &>(*promoted_type);
                const auto &right = turbo::checked_cast<const TimestampType &>(*other_type);
                if (left.timezone().empty() ^ right.timezone().empty()) {
                    return turbo::failed_precondition_error(
                            "Cannot merge timestamp with timezone and timestamp without timezone");
                }
                if (left.timezone() != right.timezone()) {
                    return turbo::failed_precondition_error("Cannot merge timestamps with differing timezones");
                }
                if (options.promote_temporal_unit) {
                    return timestamp(std::max(left.unit(), right.unit()), left.timezone());
                }
            }

            return nullptr;
        }


        std::shared_ptr<DataType> MakeBinary(const DataType &type) {
            switch (type.id()) {
                case Type::BINARY:
                case Type::STRING:
                    return binary();
                case Type::LARGE_BINARY:
                case Type::LARGE_STRING:
                    return large_binary();
                default:
                    unreachable("Hit an unknown type");
            }
            return nullptr;
        }

        // Merge temporal types based on options. Returns nullptr for non-binary types.
        turbo::Result<std::shared_ptr<DataType>> MaybeMergeBinaryTypes(
                std::shared_ptr<DataType> &promoted_type, std::shared_ptr<DataType> &other_type,
                const Field::MergeOptions &options) {
            if (options.promote_binary) {
                if (other_type->id() == Type::FIXED_SIZE_BINARY &&
                    is_base_binary_like(promoted_type->id())) {
                    return MakeBinary(*promoted_type);
                } else if (promoted_type->id() == Type::FIXED_SIZE_BINARY &&
                           is_base_binary_like(other_type->id())) {
                    return MakeBinary(*other_type);
                } else if (promoted_type->id() == Type::FIXED_SIZE_BINARY &&
                           other_type->id() == Type::FIXED_SIZE_BINARY) {
                    return binary();
                }

                if ((other_type->id() == Type::LARGE_STRING ||
                     other_type->id() == Type::LARGE_BINARY) &&
                    (promoted_type->id() == Type::STRING || promoted_type->id() == Type::BINARY)

                        ) {
                    // Promoted type is always large in case there are regular and large types
                    promoted_type.swap(other_type);
                }

                // When one field is binary and the other a string
                if (is_string(promoted_type->id()) && is_binary(other_type->id())) {
                    return MakeBinary(*promoted_type);
                } else if (is_binary(promoted_type->id()) && is_string(other_type->id())) {
                    return MakeBinary(*promoted_type);
                }

                // When the types are the same, but one is large
                if ((promoted_type->id() == Type::STRING && other_type->id() == Type::LARGE_STRING) ||
                    (promoted_type->id() == Type::LARGE_STRING && other_type->id() == Type::STRING)) {
                    return large_utf8();
                } else if ((promoted_type->id() == Type::BINARY &&
                            other_type->id() == Type::LARGE_BINARY) ||
                           (promoted_type->id() == Type::LARGE_BINARY &&
                            other_type->id() == Type::BINARY)) {
                    return large_binary();
                }
            }

            return nullptr;
        }

        // Merge list types based on options. Returns nullptr for non-list types.
        turbo::Result<std::shared_ptr<DataType>> MergeStructs(
                const std::shared_ptr<DataType> &promoted_type,
                const std::shared_ptr<DataType> &other_type, const Field::MergeOptions &options) {
            SchemaBuilder builder(SchemaBuilder::CONFLICT_APPEND, options);
            // Add the LHS fields. Duplicates will be preserved.
            TURBO_RETURN_NOT_OK(builder.add_fields(promoted_type->fields()));

            // Add the RHS fields. Duplicates will be merged, unless the field was
            // already a duplicate, in which case we error (since we don't know which
            // field to merge with).
            builder.set_policy(SchemaBuilder::CONFLICT_MERGE);
            TURBO_RETURN_NOT_OK(builder.add_fields(other_type->fields()));

            TURBO_MOVE_OR_RAISE(auto schema, builder.finish());
            return STRUCT(schema->fields());
        }

        // Merge list types based on options. Returns nullptr for non-list types.
        turbo::Result<std::shared_ptr<DataType>> MaybeMergeListTypes(
                const std::shared_ptr<DataType> &promoted_type,
                const std::shared_ptr<DataType> &other_type, const Field::MergeOptions &options) {
            if (promoted_type->id() == Type::FIXED_SIZE_LIST &&
                other_type->id() == Type::FIXED_SIZE_LIST) {
                const auto &left = turbo::checked_cast<const FixedSizeListType &>(*promoted_type);
                const auto &right = turbo::checked_cast<const FixedSizeListType &>(*other_type);
                TURBO_MOVE_OR_RAISE(
                        auto value_field,
                        left.value_field()->merge_with(
                                *right.value_field()->with_name(left.value_field()->name()), options));
                if (left.list_size() == right.list_size()) {
                    return fixed_size_list(std::move(value_field), left.list_size());
                } else {
                    return list(std::move(value_field));
                }
            } else if (is_list(promoted_type->id()) && is_list(other_type->id())) {
                const auto &left = turbo::checked_cast<const BaseListType &>(*promoted_type);
                const auto &right = turbo::checked_cast<const BaseListType &>(*other_type);
                TURBO_MOVE_OR_RAISE(
                        auto value_field,
                        left.value_field()->merge_with(
                                *right.value_field()->with_name(left.value_field()->name()), options));

                if (!options.promote_list && promoted_type->id() != other_type->id()) {
                    return turbo::failed_precondition_error("Cannot merge lists unless promote_list=true");
                }

                if (promoted_type->id() == Type::LARGE_LIST || other_type->id() == Type::LARGE_LIST) {
                    return large_list(std::move(value_field));
                } else {
                    return list(std::move(value_field));
                }
            } else if (promoted_type->id() == Type::MAP && other_type->id() == Type::MAP) {
                const auto &left = turbo::checked_cast<const MapType &>(*promoted_type);
                const auto &right = turbo::checked_cast<const MapType &>(*other_type);
                TURBO_MOVE_OR_RAISE(
                        auto key_field,
                        left.key_field()->merge_with(
                                *right.key_field()->with_name(left.key_field()->name()), options));
                TURBO_MOVE_OR_RAISE(
                        auto item_field,
                        left.item_field()->merge_with(
                                *right.item_field()->with_name(left.item_field()->name()), options));
                return map(key_field->type(), std::move(item_field),
                        /*keys_sorted=*/left.keys_sorted() && right.keys_sorted());
            } else if (promoted_type->id() == Type::STRUCT && other_type->id() == Type::STRUCT) {
                return MergeStructs(promoted_type, other_type, options);
            }

            return nullptr;
        }


        turbo::Result<std::shared_ptr<DataType>> MergeTypes(std::shared_ptr<DataType> promoted_type,
                                                            std::shared_ptr<DataType> other_type,
                                                            const Field::MergeOptions &options) {
            if (promoted_type->equals(*other_type)) return promoted_type;

            bool promoted = false;
            if (options.promote_nullability) {
                if (promoted_type->id() == Type::NA) {
                    return other_type;
                } else if (other_type->id() == Type::NA) {
                    return promoted_type;
                }
            } else if (promoted_type->id() == Type::NA || other_type->id() == Type::NA) {
                return turbo::failed_precondition_error(
                        "Cannot merge type with null unless promote_nullability=true");
            }

            if (options.promote_dictionary && is_dictionary(promoted_type->id()) &&
                is_dictionary(other_type->id())) {
                return MergeDictionaryTypes(promoted_type, other_type, options);
            }

            TURBO_MOVE_OR_RAISE(auto maybe_promoted,
                                MaybeMergeTemporalTypes(promoted_type, other_type, options));
            if (maybe_promoted) return maybe_promoted;

            TURBO_MOVE_OR_RAISE(maybe_promoted,
                                MaybeMergeNumericTypes(promoted_type, other_type, options));
            if (maybe_promoted) return maybe_promoted;

            TURBO_MOVE_OR_RAISE(maybe_promoted,
                                MaybeMergeBinaryTypes(promoted_type, other_type, options));
            if (maybe_promoted) return maybe_promoted;

            TURBO_MOVE_OR_RAISE(maybe_promoted,
                                MaybeMergeListTypes(promoted_type, other_type, options));
            if (maybe_promoted) return maybe_promoted;

            return promoted ? promoted_type : nullptr;
        }
    }

    bool Field::has_metadata() const {
        return (metadata_ != nullptr) && (metadata_->size() > 0);
    }

    std::shared_ptr<Field> Field::with_metadata(
            const std::shared_ptr<const KeyValueMetadata> &metadata) const {
        return std::make_shared<Field>(name_, type_, nullable_, metadata);
    }

    std::shared_ptr<Field> Field::with_merged_metadata(
            const std::shared_ptr<const KeyValueMetadata> &metadata) const {
        std::shared_ptr<const KeyValueMetadata> merged_metadata;
        if (metadata_) {
            merged_metadata = metadata_->Merge(*metadata);
        } else {
            merged_metadata = metadata;
        }
        return std::make_shared<Field>(name_, type_, nullable_, merged_metadata);
    }

    std::shared_ptr<Field> Field::remove_metadata() const {
        return std::make_shared<Field>(name_, type_, nullable_);
    }

    std::shared_ptr<Field> Field::with_type(const std::shared_ptr<DataType> &type) const {
        return std::make_shared<Field>(name_, type, nullable_, metadata_);
    }

    std::shared_ptr<Field> Field::with_name(const std::string &name) const {
        return std::make_shared<Field>(name, type_, nullable_, metadata_);
    }

    std::shared_ptr<Field> Field::with_nullable(const bool nullable) const {
        return std::make_shared<Field>(name_, type_, nullable, metadata_);
    }

    Field::MergeOptions Field::MergeOptions::permissive() {
        MergeOptions options = defaults();
        options.promote_nullability = true;
        options.promote_decimal = true;
        options.promote_decimal_to_float = true;
        options.promote_integer_to_decimal = true;
        options.promote_integer_to_float = true;
        options.promote_integer_sign = true;
        options.promote_numeric_width = true;
        options.promote_binary = true;
        options.promote_temporal_unit = true;
        options.promote_list = true;
        options.promote_dictionary = true;
        options.promote_dictionary_ordered = false;
        return options;
    }

    std::string Field::MergeOptions::to_string() const {
        std::stringstream ss;
        ss << "MergeOptions{";
        ss << "promote_nullability=" << (promote_nullability ? "true" : "false");
        ss << ", promote_decimal=" << (promote_decimal ? "true" : "false");
        ss << ", promote_decimal_to_float=" << (promote_decimal_to_float ? "true" : "false");
        ss << ", promote_integer_to_decimal="
           << (promote_integer_to_decimal ? "true" : "false");
        ss << ", promote_integer_to_float=" << (promote_integer_to_float ? "true" : "false");
        ss << ", promote_integer_sign=" << (promote_integer_sign ? "true" : "false");
        ss << ", promote_numeric_width=" << (promote_numeric_width ? "true" : "false");
        ss << ", promote_binary=" << (promote_binary ? "true" : "false");
        ss << ", promote_temporal_unit=" << (promote_temporal_unit ? "true" : "false");
        ss << ", promote_list=" << (promote_list ? "true" : "false");
        ss << ", promote_dictionary=" << (promote_dictionary ? "true" : "false");
        ss << ", promote_dictionary_ordered="
           << (promote_dictionary_ordered ? "true" : "false");
        ss << '}';
        return ss.str();
    }


    turbo::Result<std::shared_ptr<Field>> Field::merge_with(const Field &other,
                                                           MergeOptions options) const {
        if (name() != other.name()) {
            return turbo::invalid_argument_error("Field ", name(), " doesn't have the same name as ",
                                                 other.name());
        }

        if (equals(other, /*check_metadata=*/false)) {
            return deep_copy();
        }

        auto maybe_promoted_type = MergeTypes(type_, other.type(), options);
        if (!maybe_promoted_type.ok()) {
            return maybe_promoted_type.status().with_message(
                    "Unable to merge: Field ", name(),
                    " has incompatible types: ", type()->to_string(), " vs ", other.type()->to_string(),
                    ": ", maybe_promoted_type.status().message());
        }
        auto promoted_type = *std::move(maybe_promoted_type);
        if (promoted_type) {
            bool nullable = nullable_;
            if (options.promote_nullability) {
                nullable = nullable || other.nullable() || type_->id() == Type::NA ||
                           other.type()->id() == Type::NA;
            } else if (nullable_ != other.nullable()) {
                return turbo::failed_precondition_error("Unable to merge: Field ", name(),
                                                        " has incompatible nullability: ", nullable_, " vs ",
                                                        other.nullable());
            }

            return std::make_shared<Field>(name_, promoted_type, nullable, metadata_);
        }
        return turbo::failed_precondition_error("Unable to merge: Field ", name(),
                                                " has incompatible types: ", type()->to_string(), " vs ",
                                                other.type()->to_string());
    }


    turbo::Result<std::shared_ptr<Field>> Field::merge_with(const std::shared_ptr<Field> &other,
                                                           MergeOptions options) const {
        DKCHECK_NE(other, nullptr);
        return merge_with(*other, options);
    }

    FieldVector Field::Flatten() const {
        FieldVector flattened;
        if (type_->id() == Type::STRUCT) {
            for (const auto &child: type_->fields()) {
                auto flattened_child = child->deep_copy();
                flattened.push_back(flattened_child);
                flattened_child->name_.insert(0, name() + ".");
                flattened_child->nullable_ |= nullable_;
            }
        } else {
            flattened.push_back(this->deep_copy());
        }
        return flattened;
    }

    std::shared_ptr<Field> Field::deep_copy() const {
        return ::nebula::field(name_, type_, nullable_, metadata_);
    }

    bool Field::equals(const Field &other, bool check_metadata) const {
        if (this == &other) {
            return true;
        }
        if (this->name_ == other.name_ && this->nullable_ == other.nullable_ &&
            this->type_->equals(*other.type_.get(), check_metadata)) {
            if (!check_metadata) {
                return true;
            } else if (this->has_metadata() && other.has_metadata()) {
                return metadata_->equals(*other.metadata_);
            } else if (!this->has_metadata() && !other.has_metadata()) {
                return true;
            } else {
                return false;
            }
        }
        return false;
    }

    bool Field::equals(const std::shared_ptr<Field> &other, bool check_metadata) const {
        return equals(*other.get(), check_metadata);
    }

    bool Field::is_compatible_with(const Field &other) const { return merge_with(other).ok(); }

    bool Field::is_compatible_with(const std::shared_ptr<Field> &other) const {
        DKCHECK_NE(other, nullptr);
        return is_compatible_with(*other);
    }

    std::string Field::to_string(bool show_metadata) const {
        std::stringstream ss;
        ss << name_ << ": " << type_->to_string(show_metadata);
        if (!nullable_) {
            ss << " not null";
        }
        if (show_metadata && metadata_) {
            ss << metadata_->to_string();
        }
        return ss.str();
    }

    void print_to(const Field &field, std::ostream *os) { *os << field.to_string(); }

    // ----------------------------------------------------------------------
    // FieldPath

    size_t FieldPath::hash() const {
        return internal::ComputeStringHash<0>(indices().data(), indices().size() * sizeof(int));
    }

    std::string FieldPath::to_string() const {
        if (this->indices().empty()) {
            return "FieldPath(empty)";
        }

        std::string repr = "FieldPath(";
        for (auto index: this->indices()) {
            repr += internal::ToChars(index) + " ";
        }
        repr.back() = ')';
        return repr;
    }


    std::string Field::compute_fingerprint() const {
        const auto &type_fingerprint = type_->fingerprint();
        if (type_fingerprint.empty()) {
            // Underlying DataType doesn't support fingerprinting.
            return "";
        }
        std::stringstream ss;
        ss << 'F';
        if (nullable_) {
            ss << 'n';
        } else {
            ss << 'N';
        }
        ss << name_;
        ss << '{' << type_fingerprint << '}';
        return ss.str();
    }

    std::string Field::compute_metadata_fingerprint() const {
        std::stringstream ss;
        if (metadata_) {
            internal::append_metadata_finger_print(*metadata_, &ss);
        }
        const auto &type_fingerprint = type_->metadata_fingerprint();
        if (!type_fingerprint.empty()) {
            ss << "+{" << type_->metadata_fingerprint() << "}";
        }
        return ss.str();
    }

    // ----------------------------------------------------------------------
    // FieldRef

    FieldRef::FieldRef(FieldPath indices) : impl_(std::move(indices)) {}

    void FieldRef::Flatten(std::vector<FieldRef> children) {
        KCHECK(!children.empty());

        // flatten children
        struct Visitor {
            void operator()(std::string &&name, std::vector<FieldRef> *out) {
                out->push_back(FieldRef(std::move(name)));
            }

            void operator()(FieldPath &&path, std::vector<FieldRef> *out) {
                if (path.indices().empty()) {
                    return;
                }
                out->push_back(FieldRef(std::move(path)));
            }

            void operator()(std::vector<FieldRef> &&children, std::vector<FieldRef> *out) {
                if (children.empty()) {
                    return;
                }
                // First flatten children into temporary result
                std::vector<FieldRef> flattened_children;
                flattened_children.reserve(children.size());
                for (auto &&child: children) {
                    std::visit(std::bind(*this, std::placeholders::_1, &flattened_children),
                               std::move(child.impl_));
                }
                // If all children are FieldPaths, concatenate them into a single FieldPath
                int64_t n_indices = 0;
                for (const auto &child: flattened_children) {
                    const FieldPath *path = child.field_path();
                    if (!path) {
                        n_indices = -1;
                        break;
                    }
                    n_indices += static_cast<int64_t>(path->indices().size());
                }
                if (n_indices == 0) {
                    return;
                } else if (n_indices > 0) {
                    std::vector<int> indices(n_indices);
                    auto out_indices = indices.begin();
                    for (const auto &child: flattened_children) {
                        for (int index: *child.field_path()) {
                            *out_indices++ = index;
                        }
                    }
                    DKCHECK(out_indices == indices.end());
                    out->push_back(FieldRef(std::move(indices)));
                } else {
                    // ... otherwise, just transfer them to the final result
                    out->insert(out->end(), std::move_iterator(flattened_children.begin()),
                                std::move_iterator(flattened_children.end()));
                }
            }
        };

        std::vector<FieldRef> out;
        Visitor visitor;
        visitor(std::move(children), &out);

        if (out.empty()) {
            impl_ = std::vector<int>();
        } else if (out.size() == 1) {
            impl_ = std::move(out[0].impl_);
        } else {
            impl_ = std::move(out);
        }
    }

    turbo::Result<FieldRef> FieldRef::from_dot_path(const std::string &dot_path_arg) {
        if (dot_path_arg.empty()) {
            return FieldRef();
        }

        std::vector<FieldRef> children;

        std::string_view dot_path = dot_path_arg;

        auto parse_name = [&] {
            std::string name;
            for (;;) {
                auto segment_end = dot_path.find_first_of("\\[.");
                if (segment_end == std::string_view::npos) {
                    // dot_path doesn't contain any other special characters; consume all
                    name.append(dot_path.data(), dot_path.length());
                    dot_path = "";
                    break;
                }

                if (dot_path[segment_end] != '\\') {
                    // segment_end points to a subscript for a new FieldRef
                    name.append(dot_path.data(), segment_end);
                    dot_path = dot_path.substr(segment_end);
                    break;
                }

                if (dot_path.size() == segment_end + 1) {
                    // dot_path ends with backslash; consume it all
                    name.append(dot_path.data(), dot_path.length());
                    dot_path = "";
                    break;
                }

                // append all characters before backslash, then the character which follows it
                name.append(dot_path.data(), segment_end);
                name.push_back(dot_path[segment_end + 1]);
                dot_path = dot_path.substr(segment_end + 2);
            }
            return name;
        };

        while (!dot_path.empty()) {
            auto subscript = dot_path[0];
            dot_path = dot_path.substr(1);
            switch (subscript) {
                case '.': {
                    // next element is a name
                    children.emplace_back(parse_name());
                    continue;
                }
                case '[': {
                    auto subscript_end = dot_path.find_first_not_of("0123456789");
                    if (subscript_end == std::string_view::npos || dot_path[subscript_end] != ']') {
                        return turbo::invalid_argument_error("Dot path '", dot_path_arg,
                                                             "' contained an unterminated index");
                    }
                    children.emplace_back(std::atoi(dot_path.data()));
                    dot_path = dot_path.substr(subscript_end + 1);
                    continue;
                }
                default:
                    return turbo::invalid_argument_error("Dot path must begin with '[' or '.', got '", dot_path_arg,
                                                         "'");
            }
        }

        FieldRef out;
        out.Flatten(std::move(children));
        return out;
    }

    std::string FieldRef::to_dot_path() const {
        struct Visitor {
            std::string operator()(const FieldPath &path) {
                std::string out;
                for (int i: path.indices()) {
                    out += "[" + internal::ToChars(i) + "]";
                }
                return out;
            }

            std::string operator()(const std::string &name) { return "." + name; }

            std::string operator()(const std::vector<FieldRef> &children) {
                std::string out;
                for (const auto &child: children) {
                    out += child.to_dot_path();
                }
                return out;
            }
        };

        return std::visit(Visitor{}, impl_);
    }

    size_t FieldRef::hash() const {
        struct Visitor : std::hash<std::string> {
            using std::hash<std::string>::operator();

            size_t operator()(const FieldPath &path) { return path.hash(); }

            size_t operator()(const std::vector<FieldRef> &children) {
                size_t hash = 0;

                for (const FieldRef &child: children) {
                    hash ^= child.hash();
                }

                return hash;
            }
        };

        return std::visit(Visitor{}, impl_);
    }

    std::string FieldRef::to_string() const {
        struct Visitor {
            std::string operator()(const FieldPath &path) { return path.to_string(); }

            std::string operator()(const std::string &name) { return "Name(" + name + ")"; }

            std::string operator()(const std::vector<FieldRef> &children) {
                std::string repr = "Nested(";
                for (const auto &child: children) {
                    repr += child.to_string() + " ";
                }
                repr.resize(repr.size() - 1);
                repr += ")";
                return repr;
            }
        };

        return "FieldRef." + std::visit(Visitor{}, impl_);
    }

    std::vector<FieldPath> FieldRef::find_all(const Schema &schema) const {
        if (auto name = this->name()) {
            return internal::MapVector([](int i) { return FieldPath{i}; },
                                       schema.get_all_field_indices(*name));
        }
        return find_all(schema.fields());
    }

    std::vector<FieldPath> FieldRef::find_all(const Field &field) const {
        return find_all(field.type()->fields());
    }

    std::vector<FieldPath> FieldRef::find_all(const DataType &type) const {
        return find_all(type.fields());
    }


    struct NestedSelectorUtil {
        static turbo::Status NonStructError() {
            return turbo::unimplemented_error("get child data of non-struct array");
        }

        template<typename T>
        static const DataType *get_type(const T &input) {
            if constexpr (std::is_same_v<T, ArrayData>) {
                return input.type.get();
            } else {
                return input.type().get();
            }
        }
    };

    // Utility class for retrieving a child field/column from a top-level Field, Array,
    // ArrayData, or ChunkedArray. The "root" value can either be a single parent or a vector
    // of its children.
    template<typename T, bool IsFlattening = false>
    class NestedSelector {
    public:
        using ArrowType = T;
        using Util = NestedSelectorUtil;

        explicit NestedSelector(const std::vector<std::shared_ptr<T>> &children)
                : parent_or_children_(&children) {}

        explicit NestedSelector(const T &parent) : parent_or_children_(&parent) {}

        explicit NestedSelector(std::shared_ptr<T> parent)
                : owned_parent_(std::move(parent)), parent_or_children_(owned_parent_.get()) {}

        template<typename Arg>
        NestedSelector(Arg &&arg, MemoryPool *pool) : NestedSelector(std::forward<Arg>(arg)) {
            if (pool) {
                pool_ = pool;
            }
        }

        // If the index is out of bounds, this returns an invalid selector rather than an
        // error.
        turbo::Result<NestedSelector> GetChild(int i) const {
            std::shared_ptr<T> child;
            if (auto parent = get_parent()) {
                const DataType *type = Util::get_type(*parent);
                // We avoid this check for schema fields since it's inconsequential (plus there are
                // tests elsewhere that rely on it not happening)
                if constexpr (!std::is_same_v<T, Field>) {
                    if (TURBO_UNLIKELY(type->id() != Type::STRUCT)) {
                        return Util::NonStructError();
                    }
                }
                // Bounds-check the index *once* using the parent's type
                if (TURBO_LIKELY(i >= 0 && i < type->num_fields())) {
                    TURBO_MOVE_OR_RAISE(child, GetChild(*parent, i, pool_));
                }
            } else if (auto children = get_children()) {
                if (TURBO_LIKELY(i >= 0 && static_cast<size_t>(i) < children->size())) {
                    child = (*children)[i];
                }
            }
            return NestedSelector(std::move(child), pool_);
        }

        turbo::Result<std::shared_ptr<T>> finish() const {
            DKCHECK(get_parent() && owned_parent_);
            return owned_parent_;
        }

        template<typename OStream, typename U = T>
        std::enable_if_t<std::is_same_v<U, Field>> Summarize(OStream *os) const {
            const FieldVector *fields = get_children();
            if (!fields && get_parent()) {
                fields = &get_parent()->type()->fields();
            }
            *os << "fields: { ";
            if (fields) {
                for (const auto &field: *fields) {
                    *os << field->to_string() << ", ";
                }
            }
            *os << "}";
        }

        template<typename OStream, typename U = T>
        std::enable_if_t<!std::is_same_v<U, Field>> Summarize(OStream *os) const {
            *os << "column types: { ";
            if (auto children = get_children()) {
                for (const auto &child: *children) {
                    *os << *Util::get_type(*child) << ", ";
                }
            } else if (auto parent = get_parent()) {
                for (const auto &field: Util::get_type(*parent)->fields()) {
                    *os << *field->type() << ", ";
                }
            }
            *os << "}";
        }

        bool is_valid() const { return get_parent() || get_children(); }

        operator bool() const { return is_valid(); }

    private:
        // Accessors for the variant
        auto get_parent() const { return get_value<const T *>(); }

        auto get_children() const {
            return get_value<const std::vector<std::shared_ptr<T>> *>();
        }

        template<typename U>
        U get_value() const {
            auto ptr = std::get_if<U>(&parent_or_children_);
            return ptr ? *ptr : nullptr;
        }

        static turbo::Result<std::shared_ptr<Field>> GetChild(const Field &field, int i, MemoryPool *) {
            return field.type()->field(i);
        }

        static turbo::Result<std::shared_ptr<ArrayData>> GetChild(const ArrayData &data, int i,
                                                                  MemoryPool *pool) {
            TURBO_UNUSED(pool);
            std::shared_ptr<ArrayData> child_data;
            if constexpr (IsFlattening) {
                // First, convert to an Array so we can use StructArray::GetFlattenedField
                auto array = make_array(data.copy());
                TURBO_MOVE_OR_RAISE(auto child_array, GetChild(*array, i, pool));
                child_data = child_array->data();
            } else {
                // We could achieve the same result by converting to an Array (via make_array),
                // calling StructArray::field(i), and pulling out the new ArrayData. However, this
                // process can be very expensive when there are many columns - so we just
                // reimplement the functionality that we need
                child_data = data.child_data[i];
                if (data.offset != 0 || data.child_data[i]->length != data.length) {
                    child_data = child_data->slice(data.offset, data.length);
                }
            }

            return child_data;
        }

        static turbo::Result<std::shared_ptr<Array>> GetChild(const Array &array, int i,
                                                              MemoryPool *pool) {
            TURBO_UNUSED(pool);
            const auto &struct_array = turbo::checked_cast<const StructArray &>(array);
            if constexpr (IsFlattening) {
                return struct_array.GetFlattenedField(i, pool);
            } else {
                return struct_array.field(i);
            }
        }

        static turbo::Result<std::shared_ptr<ChunkedArray>> GetChild(const ChunkedArray &chunked_array,
                                                                     int i, MemoryPool *pool) {
            const auto &type = *chunked_array.type();

            ArrayVector chunks;
            chunks.reserve(chunked_array.num_chunks());
            for (const auto &parent_chunk: chunked_array.chunks()) {
                TURBO_MOVE_OR_RAISE(auto chunk, GetChild(*parent_chunk, i, pool));
                chunks.push_back(std::move(chunk));
            }

            return std::make_shared<ChunkedArray>(std::move(chunks), type.field(i)->type());
        }

        std::shared_ptr<T> owned_parent_;
        std::variant<const T *, const std::vector<std::shared_ptr<T>> *> parent_or_children_;
        MemoryPool *pool_ = default_memory_pool();
    };

    using FieldSelector = NestedSelector<Field>;

    template<typename T>
    using ZeroCopySelector = NestedSelector<T, false>;
    template<typename T>
    using FlatteningSelector = NestedSelector<T, true>;

    struct FieldPathGetImpl {
        template<typename Selector>
        static turbo::Status IndexError(const FieldPath *path, int out_of_range_depth,
                                        const Selector &selector) {
            std::stringstream ss;
            ss << "index out of range. ";

            ss << "indices=[ ";
            int depth = 0;
            for (int i: path->indices()) {
                if (depth++ != out_of_range_depth) {
                    ss << i << " ";
                } else {
                    ss << ">" << i << "< ";
                }
            }
            ss << "] ";

            selector.Summarize(&ss);

            return turbo::out_of_range_error(ss.str());
        }

        template<typename Selector, typename T = typename Selector::ArrowType>
        static turbo::Result<std::shared_ptr<T>> get(const FieldPath *path, Selector selector,
                                                     int *out_of_range_depth = nullptr) {
            if (path->empty()) {
                return turbo::invalid_argument_error("empty indices cannot be traversed");
            }

            int depth = 0;
            for (auto index: *path) {
                TURBO_MOVE_OR_RAISE(auto next_selector, selector.GetChild(index));

                // Handle failed bounds check
                if (!next_selector) {
                    if (out_of_range_depth) {
                        *out_of_range_depth = depth;
                        return nullptr;
                    }
                    return IndexError(path, depth, selector);
                }

                selector = std::move(next_selector);
                ++depth;
            }

            return selector.finish();
        }
    };

    std::vector<FieldPath> FieldRef::find_all(const FieldVector &fields) const {
        struct Visitor {
            std::vector<FieldPath> operator()(const FieldPath &path) {
                // skip long IndexError construction if path is out of range
                int out_of_range_depth;
                auto maybe_field =
                        FieldPathGetImpl::get(&path, FieldSelector(fields_), &out_of_range_depth);

                KCHECK_OK(maybe_field.status());

                if (maybe_field.value_or_die() != nullptr) {
                    return {path};
                }
                return {};
            }

            std::vector<FieldPath> operator()(const std::string &name) {
                std::vector<FieldPath> out;

                for (int i = 0; i < static_cast<int>(fields_.size()); ++i) {
                    if (fields_[i]->name() == name) {
                        out.push_back({i});
                    }
                }

                return out;
            }

            struct Matches {
                // referents[i] is referenced by prefixes[i]
                std::vector<FieldPath> prefixes;
                FieldVector referents;

                Matches(std::vector<FieldPath> matches, const FieldVector &fields) {
                    for (auto &match: matches) {
                        add({}, std::move(match), fields);
                    }
                }

                Matches() = default;

                [[nodiscard]] size_t size() const { return referents.size(); }

                void add(const FieldPath &prefix, const FieldPath &suffix,
                         const FieldVector &fields) {
                    auto maybe_field = suffix.get(fields);
                    KCHECK_OK(maybe_field.status());
                    referents.push_back(std::move(maybe_field).value_or_die());

                    std::vector<int> concatenated_indices(prefix.indices().size() +
                                                          suffix.indices().size());
                    auto it = concatenated_indices.begin();
                    for (auto path: {&prefix, &suffix}) {
                        it = std::copy(path->indices().begin(), path->indices().end(), it);
                    }
                    prefixes.emplace_back(std::move(concatenated_indices));
                }
            };

            std::vector<FieldPath> operator()(const std::vector<FieldRef> &refs) {
                DKCHECK_GE(refs.size(), 1);
                Matches matches(refs.front().find_all(fields_), fields_);

                for (auto ref_it = refs.begin() + 1; ref_it != refs.end(); ++ref_it) {
                    Matches next_matches;
                    for (size_t i = 0; i < matches.size(); ++i) {
                        const auto &referent = *matches.referents[i];

                        for (const FieldPath &match: ref_it->find_all(referent)) {
                            next_matches.add(matches.prefixes[i], match, referent.type()->fields());
                        }
                    }
                    matches = std::move(next_matches);
                }

                return matches.prefixes;
            }

            const FieldVector &fields_;
        };

        return std::visit(Visitor{fields}, impl_);
    }

    std::vector<FieldPath> FieldRef::find_all(const ArrayData &array) const {
        return find_all(*array.type);
    }

    std::vector<FieldPath> FieldRef::find_all(const Array &array) const {
        return find_all(*array.type());
    }

    std::vector<FieldPath> FieldRef::find_all(const ChunkedArray &chunked_array) const {
        return find_all(*chunked_array.type());
    }

    std::vector<FieldPath> FieldRef::find_all(const RecordBatch &batch) const {
        return find_all(*batch.schema());
    }

    std::vector<FieldPath> FieldRef::find_all(const Table &table) const {
        return find_all(*table.schema());
    }

    void print_to(const FieldRef &ref, std::ostream *os) { *os << ref.to_string(); }

    std::ostream &operator<<(std::ostream &os, const FieldRef &ref) {
        os << ref.to_string();
        return os;
    }

    turbo::Result<std::shared_ptr<Field>> FieldPath::get(const Schema &schema) const {
        return get(schema.fields());
    }

    turbo::Result<std::shared_ptr<Field>> FieldPath::get(const Field &field) const {
        return get(field.type()->fields());
    }

    turbo::Result<std::shared_ptr<Field>> FieldPath::get(const DataType &type) const {
        return get(type.fields());
    }

    turbo::Result<std::shared_ptr<Field>> FieldPath::get(const FieldVector &fields) const {
        return FieldPathGetImpl::get(this, FieldSelector(fields));
    }

    turbo::Result<std::shared_ptr<Schema>> FieldPath::get_all(const Schema &schm,
                                                             const std::vector<FieldPath> &paths) {
        FieldVector fields;
        fields.reserve(paths.size());
        for (const auto &path: paths) {
            TURBO_MOVE_OR_RAISE(std::shared_ptr<Field> field, path.get(schm));
            fields.push_back(std::move(field));
        }
        return schema(std::move(fields));
    }

    turbo::Result<std::shared_ptr<Array>> FieldPath::get(const RecordBatch &batch) const {
        // Deliberately calling `column_data` here because `RecordBatch::columns` is nontrivial
        TURBO_MOVE_OR_RAISE(
                auto data,
                FieldPathGetImpl::get(this, ZeroCopySelector<ArrayData>(batch.column_data())));
        return make_array(data);
    }

    turbo::Result<std::shared_ptr<ChunkedArray>> FieldPath::get(const Table &table) const {
        return FieldPathGetImpl::get(this, ZeroCopySelector<ChunkedArray>(table.columns()));
    }

    turbo::Result<std::shared_ptr<Array>> FieldPath::get(const Array &array) const {
        return FieldPathGetImpl::get(this, ZeroCopySelector<Array>(array));
    }

    turbo::Result<std::shared_ptr<ArrayData>> FieldPath::get(const ArrayData &data) const {
        return FieldPathGetImpl::get(this, ZeroCopySelector<ArrayData>(data));
    }

    turbo::Result<std::shared_ptr<ChunkedArray>> FieldPath::get(
            const ChunkedArray &chunked_array) const {
        return FieldPathGetImpl::get(this, ZeroCopySelector<ChunkedArray>(chunked_array));
    }

    turbo::Result<std::shared_ptr<Array>> FieldPath::get_flattened(const Array &array,
                                                                  MemoryPool *pool) const {
        return FieldPathGetImpl::get(this, FlatteningSelector<Array>(array, pool));
    }

    turbo::Result<std::shared_ptr<ArrayData>> FieldPath::get_flattened(const ArrayData &data,
                                                                      MemoryPool *pool) const {
        return FieldPathGetImpl::get(this, FlatteningSelector<ArrayData>(data, pool));
    }

    turbo::Result<std::shared_ptr<ChunkedArray>> FieldPath::get_flattened(
            const ChunkedArray &chunked_array, MemoryPool *pool) const {
        return FieldPathGetImpl::get(this,
                                     FlatteningSelector<ChunkedArray>(chunked_array, pool));
    }

    turbo::Result<std::shared_ptr<Array>> FieldPath::get_flattened(const RecordBatch &batch,
                                                                  MemoryPool *pool) const {
        TURBO_MOVE_OR_RAISE(
                auto data, FieldPathGetImpl::get(
                this, FlatteningSelector<ArrayData>(batch.column_data(), pool)));
        return make_array(data);
    }

    turbo::Result<std::shared_ptr<ChunkedArray>> FieldPath::get_flattened(const Table &table,
                                                                         MemoryPool *pool) const {
        return FieldPathGetImpl::get(this,
                                     FlatteningSelector<ChunkedArray>(table.columns(), pool));
    }

}  // namespace nebula